package com.xxd.dfs.datanode;

import com.xxd.dfs.namenode.rpc.model.HeartbeatRequest;
import com.xxd.dfs.namenode.rpc.model.HeartbeatResponse;
import com.xxd.dfs.namenode.rpc.model.RegisterRequest;
import com.xxd.dfs.namenode.rpc.model.RegisterResponse;
import com.xxd.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-24 11:54
 * @description:
 */
public class NameNodeRpcClient {

    private final Logger logger = LoggerFactory.getLogger(NameNodeRpcClient.class);

    private Config config = Config.getInstance();
    private Integer NAMENODE_PORT = Integer.parseInt(config.get("namenode.port"));
    private String NAMENODE_HOSTNAME = config.get("namenode.hostname");
    private String DATANODE_IP = config.get("datanode.ip");
    private String DATANODE_HOSTNAME = config.get("datanode.hostname");
    private int NIO_PORT = Integer.parseInt(config.get("nio.port"));


    private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode;

    public NameNodeRpcClient() {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();
        this.namenode = NameNodeServiceGrpc.newBlockingStub(channel);
        logger.info("与namenode hostname = {} , port = {} 建立连接", NAMENODE_HOSTNAME, NAMENODE_PORT);
    }

    /**
     * 向namenode 发起注册请求
     * @return
     */
    public boolean register() {
        logger.info("发送rpc请求 向namenode 发起注册请求.......");
        RegisterRequest request = RegisterRequest.newBuilder()
                .setIp(DATANODE_IP)
                .setHostname(DATANODE_HOSTNAME)
                .setNioPort(NIO_PORT)
                .build();

        RegisterResponse response = namenode.register(request);

        logger.info("接收到NameNode返回的注册响应：" + (response.getStatus() == 1? "SUCCESS" : "FAILURE"));
        return response.getStatus() == 1;
    }

    /**
     * 发送心跳请求
     * @return
     */
    public HeartbeatResponse heartbeat() {


        HeartbeatRequest request = HeartbeatRequest.newBuilder()
                .setIp(DATANODE_IP)
                .setHostname(DATANODE_HOSTNAME)
                .build();
        HeartbeatResponse response = namenode.heartbeat(request);

        return response;
    }
}